package job.sql.remove_duplicated;

import java.io.IOException;

import mapred.Job;
import mapred.config.CombineConfig;
import mapred.config.Compress;
import mapred.config.InputConfig;
import mapred.config.JobConfig;
import mapred.config.MapConfig;
import mapred.config.OutputConfig;
import mapred.config.ReduceConfig;
import mapred.example.RemoveDuplicatedReducer;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.TextOutputFormat;

public class Sql_RemoveDuplicated_Page extends Job {

    private int N;

    private int m;

    public Sql_RemoveDuplicated_Page(int N, int m, String... settings)
            throws IOException {
        super("Sql Remove Duplicated Page", settings);
        this.N = N;
        this.m = m;
    }

    @Override
    protected CombineConfig getCombineConfig() {
        return CombineConfig.newInstance(RemoveDuplicatedReducer.class);
    }

    @Override
    protected InputConfig getInputConfig(FileSystem fs) throws IOException {
        return InputConfig.newInstance(KeyValueTextInputFormat.class,
                "/user/guojw/SourcePage_NonDuplicated/CJK",
                "/user/guojw/SourcePage/WOW");
    }

    @Override
    protected MapConfig getMapConfig() {
        return MapConfig.newInstance(Sql_RemoveDuplicated_Page_Mapper.class,
                Text.class, Text.class);
    }

    @Override
    protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
        return OutputConfig.newInstance(TextOutputFormat.class, Text.class,
                Text.class, "SQL_RD_PAGE/Group-" + m, Compress.GZIP);
    }

    @Override
    protected ReduceConfig getReduceConfig() {
        return ReduceConfig.newInstance(RemoveDuplicatedReducer.class, 281);
    }

    @Override
    protected void setUp(JobConfig conf, FileSystem fs) throws IOException {
        conf.set("N", N);
        conf.set("m", m);
    }

    @Override
    protected void tearDown(JobConfig conf, FileSystem fs, RunningJob job)
            throws IOException {
    }

    public static void main(String[] args) throws Exception {
        // Job.buildXML(Sql_RemoveDuplicated_Page.class);

        int N = 10;
        for (int m = 0; m < N; ++m) {
            new Sql_RemoveDuplicated_Page(N, m, args).submit();
        }
    }

}
